package com.iussoft.receiveData.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.apache.tools.DateUtils;
import com.apache.tools.MD5Utils;
import com.apache.tools.StrUtil;
import com.iussoft.manager.RedisManager;
import com.iussoft.model.DataParams;
import com.iussoft.receiveData.send.SendSearchManager;
import com.iussoft.tools.JedisPoolFactory;
import com.iussoft.tools.LoggerUtil;
import com.iussoft.tools.SpringContextLoader;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.MessageListener;

import java.util.Arrays;
import java.util.List;

/**
 * @description: 异步报文 kafka消费者接受消息,topic=asyncMessage
 * @author: Hou Dayu
 * @date: Created in 2020/3/1
 */
public class AsyncKafakMessageServcie implements MessageListener<String,String> {

    private List<String> topic;
    public AsyncKafakMessageServcie(String topic){
        this.topic = Arrays.asList(topic.split(","));
    }

    @Override
    public void onMessage(ConsumerRecord<String, String> data) {
        //根据不同主题，消费
        LoggerUtil.info("AsyncKafakMessageServcie topic ="+data.topic());
        String msg = data.value();
        JSONObject json = JSON.parseObject(msg);
        String rowkey = json.getString("rowkey");
        if(StrUtil.isNull(rowkey)){
            rowkey = MD5Utils.MD5(json.getString("sendTime")+json.getString("interfaceCode"));
        }
        if(StrUtil.isNull(String.valueOf(getRedisManager().getlObjCache(rowkey, JedisPoolFactory.DATABASE)))){
            getRedisManager().createObjCache(rowkey,rowkey,0, JedisPoolFactory.DATABASE);
        }else{
            return;
        }
        DataParams params = new DataParams();
        params.setParams(json);
        params.setTraceId(rowkey);
        try{
           if(topic.contains(data.topic())){//异步报文:topic=asyncMessage
               SendSearchManager manager = (SendSearchManager) SpringContextLoader.getBean("sendSearchManager");
               if(!"1".equals(json.getString("ifTest"))) {//档板数据不在存证
                   manager.depositCertificate(params, 4);//处理信息存证
               }
               manager.asyncSendData(params);
            }
        } catch (Exception e){
            e.printStackTrace();
            LoggerUtil.errorSource("exception = Exception; method = receiveMessage; traceId = {}; requestTime = {}; requestIp = {}; sourceMessage = {}; errorMessage = {}; level = {}",params.getParamsToStr("traceId"), DateUtils.Now.fmt_yyyyMMdd_HHmmss(),params.getParamsToStr("requestUseIp"),msg,e.getMessage(),"ERROR");
        } finally {
            getRedisManager().delObjCache(rowkey, JedisPoolFactory.DATABASE);
        }
    }

    private RedisManager getRedisManager(){
        RedisManager redisManager = (RedisManager) SpringContextLoader.getBean("redisManager");
        return redisManager;
    }
}
